package cn.walls1717.wallsmall.seckill.service.impl;

import cn.walls1717.wallsmall.common.pojo.to.mq.SeckillOrderTO;
import cn.walls1717.wallsmall.common.pojo.vo.MemberRespVO;
import cn.walls1717.wallsmall.common.utils.R;
import cn.walls1717.wallsmall.seckill.feign.CouponFeignService;
import cn.walls1717.wallsmall.seckill.feign.ProductFeignService;
import cn.walls1717.wallsmall.seckill.pojo.po.SkuInfoPO;
import cn.walls1717.wallsmall.seckill.pojo.to.SeckillSkuRedisTO;
import cn.walls1717.wallsmall.seckill.pojo.vo.SeckillSessionWithSkuRelationVO;
import cn.walls1717.wallsmall.seckill.service.SeckillService;
import cn.walls1717.wallsmall.seckill.util.UserHandler;
import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static cn.walls1717.wallsmall.common.constant.order.OrderRabbitConstant.ORDER_EXCHANGE;
import static cn.walls1717.wallsmall.common.constant.order.OrderRabbitConstant.ORDER_SECKILL_ORDER_ROUTING_KEY;
import static cn.walls1717.wallsmall.seckill.constant.SeckillRedisKeyConstant.*;

/**
 * @author walls1717
 * @since 2022/11/25 16:33
 **/
@Slf4j
@Service
public class SeckillServiceImpl implements SeckillService {

    @Resource
    private CouponFeignService couponFeignService;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private ProductFeignService productFeignService;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 上架最近三天的秒杀商品<br>
     * 今天 00:00:00 - 23:59:59 <br>
     * 明天 00:00:00 - 23:59:59 <br>
     * 后天 00:00:00 - 23:59:59 <br>
     */
    @Override
    public void uploadSeckillSkuLatest3Days() {
        // 获取最近三天的秒杀场次，若方法调用失败，直接结束方法。
        R r = couponFeignService.listLatest3DaysSession();
        if (r.getCode() != 0) {
            return;
        }

        // 若最近三天无秒杀场次，直接结束方法。
        List<SeckillSessionWithSkuRelationVO> sessionWithSkuRelationVO =
            r.getData("data", new TypeReference<List<SeckillSessionWithSkuRelationVO>>() {});
        if (sessionWithSkuRelationVO == null || sessionWithSkuRelationVO.size() == 0) {
            return;
        }

        // 在数据库中保存秒杀场次与秒杀商品信息
        saveSessionsToRedis(sessionWithSkuRelationVO);
        saveSkuInfosToRedis(sessionWithSkuRelationVO);

    }

    /**
     * 获取当前时间可以参与的秒杀商品信息。<br>
     * 此方法只能获取到第一个符合条件的秒杀场次中的秒杀商品信息。
     *
     * @return 当前时间的所有秒杀商品
     */
    @Override
    public List<SeckillSkuRedisTO> listCurrentSeckillSkus() {
        log.info("-->listCurrentSeckillSkus()正常执行<--");
        long nowTime = new Date().getTime();

        try (Entry ignored = SphU.entry("seckillSkus")) {
            // 获取所有秒杀场次的key
            Set<String> keys = stringRedisTemplate.keys(SESSIONS_CACHE_PREFIX + "*");
            // Redis中没有秒杀场次数据，直接返回空。
            if (keys == null || keys.size() == 0) {
                return null;
            }

            for (String key : keys) {
                // 获取当前场次的开始时间与结束时间
                String[] timeInterval = key.replace(SESSIONS_CACHE_PREFIX, "").split("_");
                // 开始时间
                long startTime = Long.parseLong(timeInterval[0]);
                // 结束时间
                long endTime = Long.parseLong(timeInterval[1]);
                // 如果当前时间不在开始时间与结束时间的区间内，直接结束本轮循环，判断下一个秒杀场次。
                if (nowTime < startTime || nowTime > endTime) {
                    break;
                }
                // 获取当前秒杀场次中存放的skuId信息，key(秒杀场次id_skuId)，不能获取到skuId信息的直接结束循环。
                List<String> range = stringRedisTemplate.opsForList().range(key, -100, 100);
                if (range == null) {
                    break;
                }

                // 只有当前时间在秒杀场次的开始结束时间内，且秒杀场次中有skuId的才执行下面逻辑。
                BoundHashOperations<String, String, String> ops =
                    stringRedisTemplate.boundHashOps(SKU_KILL_CACHE_PREFIX);
                // 根据秒杀场次中存的 key(秒杀场次id_skuId)，获取所有秒杀商品的详细信息。
                List<String> objects = ops.multiGet(range);
                // 能获取到秒杀商品详情的则返回。
                if (objects != null) {
                    return objects.stream().map(item -> JSON.parseObject(item, SeckillSkuRedisTO.class))
                        .collect(Collectors.toList());
                }
            }
        } catch (BlockException e) {
            log.warn("-->listCurrentSeckillSkus()资源被限流<-- {}", e.getMessage());
        }

        // 所有key遍历完成还没有秒杀商品返回的直接返回空。
        return null;
    }

    /**
     * 根据skuId获取相关的秒杀信息
     *
     * @param skuId skuId
     * @return 秒杀信息
     */
    @Override
    public SeckillSkuRedisTO getSkuSeckillInfo(Long skuId) {
        BoundHashOperations<String, String, String> ops = stringRedisTemplate.boundHashOps(SKU_KILL_CACHE_PREFIX);
        Set<String> keys = ops.keys();
        if (keys == null || keys.size() == 0) {
            return null;
        }
        String regx = "\\d_" + skuId;
        for (String key : keys) {
            if (!Pattern.matches(regx, key)) {
                break;
            }
            String json = ops.get(key);
            SeckillSkuRedisTO seckillSkuRedisTO = JSON.parseObject(json, SeckillSkuRedisTO.class);
            assert seckillSkuRedisTO != null;
            Long startTime = seckillSkuRedisTO.getStartTime();
            Long endTime = seckillSkuRedisTO.getEndTime();
            long nowTime = new Date().getTime();
            if (nowTime < startTime || nowTime > endTime) {
                seckillSkuRedisTO.setRandomCode(null);
            }
            return seckillSkuRedisTO;
        }
        return null;
    }

    /**
     * 秒杀下单逻辑
     *
     * @param killId 秒杀的id(秒杀场次id_商品skuId)
     * @param key 随机码
     * @param num 购买数量
     * @return 订单号(orderSn)
     */
    @Override
    public String seckill(String killId, String key, Integer num) {
        // 从Redis中获取当前秒杀商品的数据，获取不到直接返回空。
        BoundHashOperations<String, String, String> ops = stringRedisTemplate.boundHashOps(SKU_KILL_CACHE_PREFIX);
        String skuJson = ops.get(killId);
        if (StringUtils.isEmpty(skuJson)) {
            return null;
        }

        SeckillSkuRedisTO seckillSkuRedisTO = JSON.parseObject(skuJson, SeckillSkuRedisTO.class);

        // 能获取到秒杀商品信息，判断当前时间是否在此秒杀场次的时间内，不在时间范围内则返回空。
        long nowTime = new Date().getTime();
        Long startTime = seckillSkuRedisTO.getStartTime();
        Long endTime = seckillSkuRedisTO.getEndTime();
        if (nowTime <= startTime || nowTime >= endTime) {
            return null;
        }

        // 判断随机码与killId是否与Redis中的一致，不一致直接返回空。
        String redisRandomCode = seckillSkuRedisTO.getRandomCode();
        String redisKillId = seckillSkuRedisTO.getPromotionSessionId() + "_" + seckillSkuRedisTO.getSkuId();
        if (!redisRandomCode.equals(key) || !redisKillId.equals(killId)) {
            return null;
        }

        // 当前商品购买数量大于每人限购数量，直接返回空。
        if (num > seckillSkuRedisTO.getSeckillLimit()) {
            return null;
        }

        // 如果已购买则在Redis中占位，防止重复购买。
        MemberRespVO loginUser = UserHandler.get();
        String boughtKey = loginUser.getId() + "_" + seckillSkuRedisTO.getPromotionSessionId() + killId;
        // 占位过期时间为当前秒杀场次结束
        long ttl = endTime - nowTime;
        Boolean boughtFlag =
            stringRedisTemplate.opsForValue().setIfAbsent(boughtKey, num.toString(), ttl, TimeUnit.MILLISECONDS);
        // 如果占位失败，说明该用户已经购买过，直接返回空。
        if (Boolean.FALSE.equals(boughtFlag)) {
            return null;
        }

        // 获取信号量，如果信号量能扣减成功，说明能够下单，可以执行下单逻辑。
        RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE_PREFIX + redisRandomCode);

        boolean semaphoreFlag = semaphore.tryAcquire(num);
        // 信号量扣减失败，直接返回空。
        if (!semaphoreFlag) {
            return null;
        }
        // 信号量扣减成功，执行下单逻辑。
        String orderSn = IdWorker.getTimeId();
        SeckillOrderTO seckillOrderTO = new SeckillOrderTO();
        seckillOrderTO.setOrderSn(orderSn);
        seckillOrderTO.setMemberId(loginUser.getId());
        seckillOrderTO.setSkuId(seckillSkuRedisTO.getSkuId());
        seckillOrderTO.setNum(num);
        seckillOrderTO.setPromotionSessionId(seckillSkuRedisTO.getPromotionSessionId());
        seckillOrderTO.setSeckillPrice(seckillSkuRedisTO.getSeckillPrice());
        rabbitTemplate.convertAndSend(ORDER_EXCHANGE, ORDER_SECKILL_ORDER_ROUTING_KEY, seckillOrderTO);
        return orderSn;

    }

    /**
     * 在Redis中保存秒杀场次信息<br>
     * key：seckill:sessions:开始时间_结束时间<br>
     * value：秒杀场次promotionSessionId_参与秒杀商品的skuId<br>
     *
     * @param sessionWithSkuRelationVO sessionWithSkuRelationVO
     */
    private void saveSessionsToRedis(List<SeckillSessionWithSkuRelationVO> sessionWithSkuRelationVO) {
        sessionWithSkuRelationVO.forEach(session -> {
            // 秒杀开始时间
            long startTime = session.getStartTime().getTime();
            // 秒杀结束时间
            long endTime = session.getEndTime().getTime();
            // seckill:sessions:开始时间_结束时间
            String key = SESSIONS_CACHE_PREFIX + startTime + "_" + endTime;
            // 若Redis中已经存过当前场次，则无需再存。
            Boolean hasKey = stringRedisTemplate.hasKey(key);
            if (Boolean.FALSE.equals(hasKey)) {
                List<String> skuIds = session.getRelationSkus().stream()
                    .map(relation -> relation.getPromotionSessionId() + "_" + relation.getSkuId().toString())
                    .collect(Collectors.toList());
                stringRedisTemplate.opsForList().leftPushAll(key, skuIds);
            }
        });
    }

    /**
     * 将秒杀商品信息保存到Redis，同时使用分布式信号量达到限流目的<br>
     * 秒杀商品<br>
     * key：seckill:skus 秒杀场次promotionSessionId_参与秒杀商品的skuId<br>
     * value：seckillSkuRedisTO<br>
     * 库存信号量<br>
     * key：seckill:stock:随机码<br>
     * value：库存数量
     * 
     * @param sessionWithSkuRelationVO sessionWithSkuRelationVO
     */
    private void saveSkuInfosToRedis(List<SeckillSessionWithSkuRelationVO> sessionWithSkuRelationVO) {
        sessionWithSkuRelationVO.forEach(session -> {

            // 绑定hash前缀 seckill:skus
            BoundHashOperations<String, Object, Object> ops = stringRedisTemplate.boundHashOps(SKU_KILL_CACHE_PREFIX);

            // 处理秒杀商品数据
            session.getRelationSkus().forEach(relation -> {
                // 判断key 秒杀场次promotionSessionId_参与秒杀商品的skuId是否存在，不存在则执行保存逻辑。
                Boolean hasSkuKey =
                    ops.hasKey(relation.getPromotionSessionId().toString() + "_" + relation.getSkuId().toString());
                if (Boolean.FALSE.equals(hasSkuKey)) {

                    SeckillSkuRedisTO seckillSkuRedisTO = new SeckillSkuRedisTO();
                    BeanUtils.copyProperties(relation, seckillSkuRedisTO);

                    // 获取sku的详细信息，获取成功则设置sku信息。
                    R r = productFeignService.getSkuInfoBySkuId(relation.getSkuId());
                    if (r.getCode() == 0) {
                        SkuInfoPO skuInfo = r.getData("skuInfo", new TypeReference<SkuInfoPO>() {});
                        seckillSkuRedisTO.setSkuInfo(skuInfo);
                    }

                    // 设置秒杀的开始时间和结束时间
                    seckillSkuRedisTO.setStartTime(session.getStartTime().getTime());
                    seckillSkuRedisTO.setEndTime(session.getEndTime().getTime());

                    // 设置随机码
                    String randomCode = UUID.randomUUID().toString().replace("-", "");
                    seckillSkuRedisTO.setRandomCode(randomCode);

                    // 保存sku秒杀数据到Redis
                    String jsonString = JSON.toJSONString(seckillSkuRedisTO);
                    ops.put(relation.getPromotionSessionId().toString() + "_" + relation.getSkuId().toString(),
                        jsonString);

                    // 设置库存信号量，起到限流作用。
                    RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE_PREFIX + randomCode);
                    semaphore.trySetPermits(relation.getSeckillCount());
                }

            });

        });
    }
}
